package com.hs.demo;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.debezium.connector.mysql.MySqlConnector;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import io.debezium.relational.history.FileDatabaseHistory;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.storage.FileOffsetBackingStore;

import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
public class Demo01 {
    public static void main(String[] args) {
        Properties props = genProps(); // 0. 配置数据库，添加用户，赋予主从同步的权限  // 1. 生成配置
        DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)  // 2. 构建 DebeziumEngine    // 使用 Json 格式
                .using(props)
                .notifying(record -> {
                    log.error("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~>>>");
//                    System.out.println(record);
                    String value = record.value();
                    JSONObject jsonObject = JSON.parseObject(value);
                    Object act = jsonObject.get("op");
                    Object source = jsonObject.get("source");
                    JSONObject jsonObjectSource = JSON.parseObject(source.toString());
                    Object table = jsonObjectSource.get("table");
                    log.error(table.toString());
                    if(act.equals("u")){
                        log.error("更新操作");
                    }else if(act.equals("d")){
                        log.error("删除");
                    }else if(act.equals("c")){
                        log.error("新增");
                    }
                    log.error("record.key() = " + record.key());// record中会有操作的类型（增、删、改）和具体的数据  // key是主键
                    log.error("record.value() = " + jsonObject);
                    log.error("<<<~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
                })
                .using((success, message, error) -> { // 强烈建议加上此部分的回调代码，方便查看错误信息
                    if (!success && error != null) {
                        log.error("----------error------");  // 报错回调
                        log.info(message);
                    }else {
                        log.info(message);
                    }
                }).build();
        ExecutorService executor = Executors.newSingleThreadExecutor();// 3. 正式运行
        executor.execute(engine);
    }
    // 配置
    private static Properties genProps() {
        Properties props = new Properties();
        // 在maven处引入其他数据库的连接器，例如debezium-connector-postgres，再修改此处的connector.class，即可使用其他数据库的CDC
        props.setProperty("connector.class", MySqlConnector.class.getCanonicalName());
        props.setProperty("database.server.name", "md_mes_db"); // 可以任意修改
        props.setProperty("database.hostname", "localhost"); // IP
        props.setProperty("database.port","3306"); // 端口
        props.setProperty("database.user", "root"); // 用户
        props.setProperty("database.password", "root"); // 密码
        props.setProperty("database.serverTimezone", "UTC"); // 时区
        // 下面两个是数据库和表，注意只能选择一种:
        // 1. 使用database.whitelist，只设置数据库（会通知全库的CDC信息）
        // 2. 使用table.whitelist，设置库名和表名（会通知单个库的单个表的CDC信息）
//        props.setProperty("database.whitelist", "db_inventory_cdc");
        props.setProperty("table.whitelist", "md_mes_db.tb_upload_pouringpack"); // 库.表名
        props.setProperty("name", "engine");
        props.setProperty("key.converter.schemas.enable", "false");
        props.setProperty("value.converter.schemas.enable", "false");
        props.setProperty("include.schema.changes", "false");
        props.setProperty("tombstones.on.delete", "false");
        props.setProperty("database.history", FileDatabaseHistory.class.getCanonicalName());
        props.setProperty("database.history.store.only.monitored.tables.ddl", "true");
        props.setProperty("database.history.file.filename", "E://debezuim/history.dat");
        props.setProperty("database.history.instance.name", UUID.randomUUID().toString());
        props.setProperty("database.history.skip.unparseable.ddl", "true");
        props.setProperty("offset.storage", FileOffsetBackingStore.class.getCanonicalName());// 偏移量持久化配置
        props.setProperty("offset.storage.file.filename", "E://debezuim/offsets.dat");
        props.setProperty("offset.flush.interval.ms", "1");
        return props;
    }





}
